#pragma once

#include <fstream>

#include <butil/string_printf.h>

#include "index/attribute_index/attribute_disk_indexer/attribute_disk_indexer.h"

namespace hawking {
namespace indexlib {

template<class T>
class MultiValueAttributeDiskIndexer : public AttributeDiskIndexer {
public:
    static const uint32_t item_offset_length_ = sizeof(uint64_t);
    static const uint32_t elem_count_length_ = sizeof(ArrayLengthType);
    static const uint32_t item_length_ = item_offset_length_ + elem_count_length_;

    MultiValueAttributeDiskIndexer(
        int32_t field_id,
        util::ColumnType field_type,
        std::shared_ptr<AttributeConvertor> attribute_convertor) : 
            AttributeDiskIndexer(field_id, field_type, attribute_convertor) {}
    virtual ~MultiValueAttributeDiskIndexer() = default;
    
    virtual Status BatchRead(SegmentContext* segment_ctx) override;

    uint32_t ElementLength() const override {
        return sizeof(T);
    }

    virtual uint64_t DocCount() const override {
        return offset_reader_ ? (offset_reader_->Length() / sizeof(size_t)) : 0;
    }

    virtual util::HawkingRecallRetCode UpdateField_(
        DocId doc_id, const std::string_view& value) override;

    virtual bool IsVarlen() const {
        return true;
    }

private:
    Status IncreaseRead_(DocId doc_id, size_t offset_in_sra, util::FieldValue* field_value);
    Status MakeFilePart_(
        size_t offset_in_offsetfile,
        util::FieldValue* field_value,
        std::vector<FilePart>* immutable_datafile_part);

    Status ReadElemCountAndDataFileOffset_(
        size_t offset_in_offsetfile,
        size_t* datafile_offset,
        ArrayLengthType* elem_count) const;
    void MakeFilePartOfCharTyped_(
        util::FieldValue* field_value,
        ArrayLengthType elem_count,
        size_t datafile_offset) const;
    void MakeFilePartOfIntTyped_(
        util::FieldValue* field_value,
        ArrayLengthType elem_count,
        size_t datafile_offset) const;
    void MakeFilePartOfFloatTyped_(
        util::FieldValue* field_value,
        ArrayLengthType elem_count,
        size_t datafile_offset) const;
    void MakeFilePartOfStringTyped_() const {}
    void FloatListPostDispose_(SegmentContext* segment_ctx);

    size_t ReadOffsetInVarlenMetaFile_(DocId doc_id) const {
        size_t offset_in_offsetfile;
        off_t offset_in_offsetfile = doc_id * sizeof(offset_in_offsetfile);
        offset_reader_->Read(
            (uint8_t*)(&offset_in_offsetfile),
            sizeof(offset_in_offsetfile),
            offset_in_offsetfile);
        return offset_in_offsetfile;
    }
};

template<>
inline uint32_t MultiValueAttributeDiskIndexer<std::string>::ElementLength() const {
    return 0;
}

template<class T>
util::HawkingRecallRetCode MultiValueAttributeDiskIndexer<T>::UpdateField_(
    DocId doc_id, const std::string_view& value) {
    if (unlikely(doc_id >= docs_updator_.size())) {
        return util::HawkingRecallRetCode::INVALID_ARGS;
    }

    if (!docs_updator_[doc_id]) {
        docs_updator_[doc_id] = std::make_unique<SimpleRecycleArray>(
            slot_count_for_update_, slot_bytes_for_update_);
    }

    size_t offset = docs_updator_[doc_id]->Insert(value.data(), value.length());
    if (-1 == offset) {
        return util::HawkingRecallRetCode::ATTR_UPDATE_FAIL;
    }

    // write `datafile-length + offset` to offset file
    offset += offset_reader_->Length();
    size_t offset_file_offset = doc_id * sizeof(size_t);
    auto ret = offset_reader_->Write(
        (const uint8_t*)&offset, sizeof(offset), offset_file_offset);
    if (!ret.OK()) {
        return util::HawkingRecallRetCode::OFFSET_UPDATE_FAIL;
    }

    return util::HawkingRecallRetCode::OK;
}

template<class T>
Status MultiValueAttributeDiskIndexer<T>::ReadElemCountAndDataFileOffset_(
    size_t offset_in_offsetfile,
    size_t* datafile_offset,
    ArrayLengthType* elem_count) const {
    auto ret = varlen_metafile_reader_->Read(
        (uint8_t*)datafile_offset, sizeof(*datafile_offset), offset_in_offsetfile);
    if (unlikely(!ret.OK())) {
        LOG(ERROR) << "read varlen_metafile fail with offset " << offset_in_offsetfile;
        return ret.Code();
    }

    offset_in_offsetfile += sizeof(*datafile_offset);
    ret = varlen_metafile_reader_->Read(
        (uint8_t*)elem_count, sizeof(*elem_count), offset_in_offsetfile);
    if (unlikely(!ret.OK())) {
        LOG(ERROR) << "read varlen_metafile fail with offset " << offset_in_offsetfile;
        return ret.Code();
    }

    return Status::OK;
}

template<class T>
Status MultiValueAttributeDiskIndexer<T>::BatchRead(SegmentContext* segment_ctx) {
    Status ret = Status::OK;
    std::vector<FilePart> immutable_datafile_part;
    std::vector<DocContext>& doc_context = *(segment_ctx->MutableDocContext());
    for (size_t idx = 0; idx < segment_ctx->DocIds().size(); ++idx) {
        if (segment_ctx->DocContexts()[idx].filtered) {
            continue;
        }

        util::FieldValue* field_value =
            segment_ctx->ArenaCtx().CreateOnArena<util::FieldValue>();
        doc_context[idx].field2attr.emplace(field_id_, field_value);

        size_t offset_in_offsetfile =
            ReadOffsetInVarlenMetaFile_(segment_ctx->DocIds()[idx]);
        if (IsUpdated_(offset_in_offsetfile)) {
            ret = IncreaseRead_(
                segment_ctx->DocIds()[idx],
                offset_in_offsetfile - offset_reader_->Length(),
                field_value);
            if (ret != Status::OK) {
                return ret;
            }
        } else {
            ret = MakeFilePart_(
                offset_in_offsetfile, field_value, &immutable_datafile_part);
            if (ret != Status::OK) {
                return ret;
            }
        }
    }

    if (!immutable_datafile_part.empty() &&
        (ret = data_reader_->BatchRead(immutable_datafile_part).Code()) != Status::OK) {
        return ret;
    }

    if (field_type_ == util::ColumnType::COLUMN_FLOAT_LIST) {
        FloatListPostDispose_(segment_ctx);
    }

    return Status::OK;
}

template<class T>
void MultiValueAttributeDiskIndexer<T>::FloatListPostDispose_(SegmentContext* segment_ctx) {
    std::vector<DocContext>& doc_ctxs = *(segment_ctx->MutableDocContext());
    for (auto& doc_ctx : doc_ctxs) {
        util::FieldValue* field_value = doc_ctx.GetField(field_id_);
        for (int32_t edx = 0; edx < field_value->double_values().values_size(); ++edx) {
            float* f_ptr = reinterpret_cast<float*>(
                field_value->mutable_double_values()->mutable_values()->Mutable(edx));
            field_value->mutable_double_values()->mutable_values()->Set(edx, *f_ptr);
        }
    }
}

template<class T>
void MultiValueAttributeDiskIndexer<T>::MakeFilePartOfCharTyped_(
    ArrayLengthType elem_count,
    size_t datafile_offset,
    util::FieldValue* field_value,
    std::vector<FilePart>* immutable_datafile_part) const {
    field_value->mutable_bytes_value()->resize(elem_count);
        
    immutable_datafile_part->emplace_back(
        datafile_offset,
        elem_count,
        reinterpret_cast<uint8_t*>(field_value->mutable_bytes_value()->data()));
}

template<class T>
void MultiValueAttributeDiskIndexer<T>::MakeFilePartOfIntTyped_(
    ArrayLengthType elem_count,
    size_t datafile_offset,
    util::FieldValue* field_value,
    std::vector<FilePart>* immutable_datafile_part) const {
    immutable_datafile_part->reserve(elem_count + immutable_datafile_part->size());
    field_value->mutable_int_values()->mutable_values()->Resize(elem_count, 0);
    for (size_t elem_idx = 0; elem_idx < elem_count; ++elem_idx) {
        int64_t* p_int64 =
            field_value->mutable_int_values()->mutable_values()->Mutable(elem_idx);
        *p_int64 = 0;
        
        immutable_datafile_part->emplace_back(
            datafile_offset,
            ElementLength(),
            reinterpret_cast<uint8_t*>(p_int64));
        datafile_offset += ElementLength();
    }
}

template<class T>
void MultiValueAttributeDiskIndexer<T>::MakeFilePartOfFloatTyped_(
    ArrayLengthType elem_count,
    size_t datafile_offset,
    util::FieldValue* field_value,
    std::vector<FilePart>* immutable_datafile_part) const {
    immutable_datafile_part->reserve(elem_count + immutable_datafile_part->size());
    field_value->mutable_double_values()->mutable_values()->Resize(elem_count, 0);
    for (size_t elem_idx = 0; elem_idx < elem_count; ++elem_idx) {
        double* p_int64 =
            field_value->mutable_double_values()->mutable_values()->Mutable(elem_idx);
        *p_int64 = 0;
        
        immutable_datafile_part->emplace_back(
            datafile_offset,
            ElementLength(),
            reinterpret_cast<uint8_t*>(p_int64));
        datafile_offset += ElementLength();
    }
}

template<>
void MultiValueAttributeDiskIndexer<std::string>::MakeFilePartOfStringTyped_(
    ArrayLengthType elem_count,
    size_t datafile_offset,
    const std::vector<StringLengthType>& string_lengths,
    util::FieldValue* field_value,
    std::vector<FilePart>* immutable_datafile_part) const {
    immutable_datafile_part->reserve(elem_count + immutable_datafile_part->size());
    field_value->mutable_bytes_values()->mutable_values()->Reserve(elem_count);
    for (size_t elem_idx = 0; elem_idx < elem_count; ++elem_idx) {
        std::string* pstr = field_value->mutable_bytes_values()->add_values();
        pstr->resize(string_lengths[elem_idx]);

        immutable_datafile_part->emplace_back(
            datafile_offset,
            string_lengths[elem_idx],
            reinterpret_cast<uint8_t*>(pstr->data()));
        datafile_offset += string_lengths[elem_idx];
    }
}

template<class T>
Status MultiValueAttributeDiskIndexer<T>::MakeFilePart_(
    size_t offset_in_offsetfile,
    util::FieldValue* field_value,
    std::vector<FilePart>* immutable_datafile_part) {
    // 1. read offset_in_datafile + elem_count from varlen_metafile
    size_t datafile_offset;
    ArrayLengthType elem_count;
    Status ret = ReadElemCountAndDataFileOffset_(
        offset_in_offsetfile, &datafile_offset, &elem_count);
    if (unlikely(Status::OK != ret)) {
        return ret;
    }

    // 2. makeup filepart for every elem
    if (field_type_ == util::ColumnType::COLUMN_INT8_LIST ||
        field_type_ == util::ColumnType::COLUMN_INT16_LIST ||
        field_type_ == util::ColumnType::COLUMN_INT32_LIST ||
        field_type_ == util::ColumnType::COLUMN_INT64_LIST) {
        MakeFilePartOfIntTyped_(
            elem_count, datafile_offset, field_value, immutable_datafile_part);
    } else if (field_type == util::ColumnType::COLUMN_FLOAT_LIST ||
               field_type == util::ColumnType::COLUMN_DOUBLE_LIST) {
        MakeFilePartOfFloatTyped_(
            elem_count, datafile_offset, field_value, immutable_datafile_part);
    } else if (field_type == util::ColumnType::COLUMN_STRING) {
        MakeFilePartOfCharTyped_(
            elem_count, datafile_offset, field_value, immutable_datafile_part);
    }

    return Status::OK;
}

// multi string
template<>
Status MultiValueAttributeDiskIndexer<std::string>::MakeFilePart_(
    size_t offset_in_offsetfile,
    util::FieldValue* field_value,
    std::vector<FilePart>* immutable_datafile_part) {
    // 1. read offset_in_datafile + elem_count from varlen_metafile
    size_t datafile_offset;
    ArrayLengthType elem_count;
    Status ret = ReadElemCountAndDataFileOffset_(
        offset_in_offsetfile, &datafile_offset, &elem_count);
    if (unlikely(Status::OK != ret)) {
        return ret;
    }

    offset_in_offsetfile += sizeof(offset_in_offsetfile);
    offset_in_offsetfile += sizeof(elem_count);
    std::vector<StringLengthType> string_lengths(elem_count, 0);
    if (elem_count) {
        auto r = varlen_metafile_reader_->Read(
            (uint8_t*)(string_lengths.data()),
            elem_count * sizeof(StringLengthType),
            offset_in_offsetfile);
        if (unlikely(!r.OK())) {
            LOG(ERROR) << "read varlen_metafile fail with offset "
                << offset_in_offsetfile;
            return r.Code();
        }
    }

    // 2. makeup filepart for every string elem
    MakeFilePartOfStringTyped_(
        elem_count, datafile_offset, string_lengths,
        field_value, immutable_datafile_part);
    
    return Status::OK;
}

template<class T>
Status MultiValueAttributeDiskIndexer<T>::IncreaseRead_(
    DocId doc_id, size_t offset_in_sra, util::FieldValue* field_value) {
    if (unlikely(doc_id >= docs_updator_.size() || docs_updator_[doc_id] == nullptr)) {
        LOG(ERROR) << "doc_id " << doc_id << "'s sra is nullptr";
        return Status::NotFound;
    }

    int8_t* sra_addr = docs_updator_[doc_id]->Get(offset_in_sra);
    if (unlikely(!sra_addr)) {
        LOG(ERROR) << "doc_id " << doc_id << " sra_addr is nullptr";
        return Status::InternalError;
    }

    attribute_convertor_->Decode(sra_addr, field_value);
    docs_updator_[doc_id]->Release(offset_in_sra);
    return Status::OK;
}

}
}